package com.atguigu.edu.app.dwd.db;

import com.atguigu.edu.app.BaseSQLApp;
import com.atguigu.edu.common.Constant;
import com.atguigu.edu.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.hadoop.yarn.util.StringHelper.join;

public class Dwd_04_TradePayDetailSuc extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_04_TradePayDetailSuc().init(
                3004,2,
                "Dwd_04_TradePayDetailSuc"
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // 1. 读取 ods_db 数据
        readOdsDb(tEnv, "Dwd_04_TradePayDetailSuc");

        // 2. 过滤出所需数据
        filterOdsData(tEnv);

        // 3.读取 Order_Detail
        readOrderDetail(tEnv);

        // 4.表join
        join(tEnv);

        // 5.写出到kafka
        writeToKafka(tEnv);
    }


    private void writeToKafka(StreamTableEnvironment tEnv) {
        tEnv.executeSql("create table dwd_trade_pay_detail_suc(" +
                "order_detail_id string," +
                "order_id string," +
                "user_id string," +
                "course_id string," +
                "course_name string," +
                "province_id string," +
                "date_id string," +
                "callback_time string," +
                "payment_type string," +
                "source_id string," +
                "origin_amount string," +
                "coupon_reduce string," +
                "final_amount string," +
                "ts bigint " +
                ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_PAY_DETAIL_SUC));

        tEnv.executeSql("insert into dwd_trade_pay_detail_suc select * from join_result");
    }

    private void join(StreamTableEnvironment tEnv) {

        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60 + 5));

        Table joinResult = tEnv.sqlQuery("select " +
                "od.id order_detail_id," +
                "od.order_id," +
                "od.user_id," +
                "od.course_id," +
                "od.course_name," +
                "od.province_id," +
                "pi.date_id," +
                "pi.callback_time," +
                "pi.payment_type," +
                "od.source_id," +
                "od.origin_amount," +
                "od.coupon_reduce," +
                "od.final_amount," +
                "pi.ts " +
                "from payment_info pi " +
                "join dwd_trade_order_detail od on pi.order_id=od.order_id " );

//        joinResult.execute().print();
        tEnv.createTemporaryView("join_result",joinResult);
    }

    private void readOrderDetail(StreamTableEnvironment tEnv) {
        // 读取下单事务事实表
        tEnv.executeSql("create table dwd_trade_order_detail(" +
                "id string," +
                "order_id string," +
                "user_id string," +
                "course_id string," +
                "course_name string," +
                "province_id string," +
                "date_id string," +
                "create_time string," +
                "source_id string," +
                "origin_amount string," +
                "coupon_reduce string," +
                "final_amount string," +
                "ts bigint)" + SQLUtil.getKafkaSourceDDL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, "Dwd_04_TradePayDetailSuc"));
    }

    private void filterOdsData(StreamTableEnvironment tEnv) {
        // 过滤支付成功数据
        Table paymentInfo = tEnv.sqlQuery("select " +
                "data['order_id'] order_id," +
                "data['payment_type'] payment_type," +
                "data['callback_time'] callback_time," +
                "date_format(`data`['callback_time'],'yyyy-MM-dd') date_id," +
                "ts " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='payment_info' " +
                "and `data`['callback_time'] is not null ");

//        paymentInfo.execute().print();
        tEnv.createTemporaryView("payment_info", paymentInfo);
    }
}
